package com.meow_woof.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.meow_woof.dto.Result;
import com.meow_woof.entity.VoucherOrder;
import com.meow_woof.mapper.VoucherOrderMapper;
import com.meow_woof.service.ISeckillVoucherService;
import com.meow_woof.service.IVoucherOrderService;
import com.meow_woof.utils.RedisIdWorker;
import com.meow_woof.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 服务实现类
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedissonClient redissonClient;

    private IVoucherOrderService proxy;

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT=new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }
    //private final BlockingQueue<VoucherOrder> orderTasks=new ArrayBlockingQueue<>(1024*1024);
    private static final ExecutorService SECKILL_ORDER_EXECUTOR=Executors.newSingleThreadExecutor();
    @PostConstruct
    private void init(){
        log.info("init");
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());

    }private class VoucherOrderHandler implements Runnable{
        /**
         * Runs this operation.
         */
        public void run() {
            while (true){
                try {
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    if(list==null||list.isEmpty()){
                        continue;
                    }
                    MapRecord<String, Object, Object> record = list.getFirst();
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);



                    handleVoucherOrder(voucherOrder);
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders","g1",record.getId());

                } catch (Exception e) {
                    log.error("fail");
                    try {
                        handlePendingList();
                    } catch (InterruptedException ex) {
                        log.error("fail");
                    }
                }
            }
        }
    }

    private void handlePendingList() throws InterruptedException {
        while (true){
            try {
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                );
                if(list==null||list.isEmpty()){
                    break;
                }
                MapRecord<String, Object, Object> record = list.getFirst();
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);



                handleVoucherOrder(voucherOrder);
                stringRedisTemplate.opsForStream().acknowledge("stream.orders","g1",record.getId());

            } catch (Exception e) {
                log.error("fail");
                Thread.sleep(20);
            }
        }
    }

    /*private class VoucherOrderHandler implements Runnable {
        *//**
         * Runs this operation.
         *//*
        public void run() {
             while (true) {

                try {
                    VoucherOrder voucherOrder = orderTasks.take();
                    handleVoucherOrder(voucherOrder);

                } catch (Exception e) {
                    log.error("处理订单异常", e);
                }


            }
        }
    }*/


    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();

        RLock lock = redissonClient.getLock("lock:order" + userId);
        boolean isLock = lock.tryLock();
        if(!isLock){
            return ;

        }


        try {

            proxy.createVoucherOrder(voucherOrder);
        }
        finally{
            lock.unlock();

        }
    }
    public Result seckillVocher(Long voucherId) {
        long orderId = redisIdWorker.nextId("order");
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), UserHolder.getUser().getId().toString(),String.valueOf(orderId));
        int r=result.intValue();
        if(r!=0){
            return Result.fail(r==1?"库存不足":"一个用户只能买一单");
        }

    //TODO
        proxy= (IVoucherOrderService) AopContext.currentProxy();




        return Result.ok(orderId);
    }



    /*public Result seckillVocher(Long voucherId) {
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), UserHolder.getUser().getId().toString());
        int r=result.intValue();
        if(r!=0){
            return Result.fail(r==1?"库存不足":"一个用户只能买一单");
        }
        long orderId = redisIdWorker.nextId("order");
        long userId=UserHolder.getUser().getId();
        VoucherOrder voucherOrder = VoucherOrder.builder()
                .id(orderId)
                .userId(userId)
                .voucherId(voucherId)
                .build();
        log.info("成功加入阻塞队列：{}", voucherOrder);

        orderTasks.add(voucherOrder);
        proxy= (IVoucherOrderService) AopContext.currentProxy();
        orderTasks.add(voucherOrder);
        log.info("队列当前大小: {}", orderTasks.size());

        log.info("proxy：{}", proxy);


        return Result.ok(orderId);
    }*/

    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        Long voucherId = voucherOrder.getVoucherId();
        Long userId = voucherOrder.getUserId();


        long count=query().eq("voucher_id",voucherId).eq("user_id",userId).count();
        if(count>0){
            return ;
        }

        boolean success = seckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherId).gt("stock",0).update();
        if(!success){
            return ;

        }
        save(voucherOrder);


    }
}
